package com.webchat.pgc.service.mall;


import com.webchat.common.enums.LUAScriptConstants;
import com.webchat.common.enums.MessageQueueTaskStatusEnum;
import com.webchat.common.enums.RedisKeyEnum;
import com.webchat.common.enums.messagequeue.MessageQueueEnum;
import com.webchat.common.enums.messagequeue.MessageQueueMethodEnum;
import com.webchat.common.exception.BusinessException;
import com.webchat.common.service.RedisService;
import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
import com.webchat.common.service.messagequeue.task.MessageQueueTaskDTO;
import com.webchat.common.util.JsonUtil;
import com.webchat.domain.dto.queue.MallProductStockUpdateMessageDTO;
import com.webchat.pgc.messagequeue.local.MQTaskWriteDataBaseQueue;
import com.webchat.pgc.messagequeue.task.PgcMessageQueueTaskService;
import com.webchat.pgc.repository.dao.mall.IMallSKUDAO;
import com.webchat.pgc.repository.entity.mall.MallSKU;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class MallProductStockService {


    @Autowired
    private IMallSKUDAO mallSkuDAO;

    @Autowired
    private RedisService redisService;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private MessageQueueProducer<MallProductStockUpdateMessageDTO, Integer> messageQueueProducer;

    @Autowired
    private MQTaskWriteDataBaseQueue taskWriteDataBaseQueue;

    /**
     * 代表缓存不存在
     */
    private static final String CACHE_NOT_EXIST = "-2";
    /**
     * 库存在执行后将被扣除为负数
     */
    private static final String CACHE_NOT_ENOUGH = "-1";
    /**
     * 更新失败
     */
    private static final String CACHE_UPDATE_ERROR = "0";
    /**
     * 库存更新成功
     */
    private static final String CACHE_UPDATE_SUCCESS = "1";


    /**
     * 库存扣减
     *
     * 注：这里需要关注并发场景，”防超卖“
     *
     * @param spuId
     * @param skuId
     * @param num
     * @return
     */
    public boolean reduceStock(Long spuId, Long skuId, Integer num) {

        return incrementStock(spuId, skuId, -1 * num);
    }

    /**
     * 库存新增
     *
     * @param skuId
     * @param num
     * @return
     */
    public boolean incrementStock(Long spuId, Long skuId, Integer num) {
        String stockRedisCacheKey = getProductStockCacheKey(spuId);

        // 基于LUA保证库存redis数据读写原子性（读、计算、写）
        String execRes = redisService.executeScript(RedisScript.of(LUAScriptConstants.UPDATE_SKU_STOCK_LUA, String.class),
                List.of(stockRedisCacheKey),
                String.valueOf(skuId), String.valueOf(num));
        if (StringUtils.isBlank(execRes)) {
            throw new BusinessException("库存服务异常");
        }
        // 处理执行结果
        if (CACHE_UPDATE_SUCCESS.equals(execRes)) {

            // redis 库存量更新更新成功，需要更新数据库库存字段
            // 更新策略：
            //   将库存更新任务丢到普通MQ，排队消费更新数据库（延迟更新，保证数据库与redis库存量最终一致）
            MallProductStockUpdateMessageDTO message = buildQueueMessage(skuId, num);
            try {
                pushProductStockUpdateMessageQueue(message);
            } catch (Exception e) {
                // 分布式队列入队异常，失败任务保存到数据库，后续有MQ失败任务调度重新归对
                // com.webchat.pgc.config.ScheduleConfig.handleFailMQTaskSchedule
                MessageQueueTaskDTO taskDTO = buildMessageQueueTaskDTO(message);
                // 放入本地队列
                try {
                    taskWriteDataBaseQueue.push(taskDTO);
                } catch (Exception pushException) {
                    rollbackRedisStock(spuId, skuId, num);
                    return false;
                }
            }
            return true;
        } else if (CACHE_UPDATE_ERROR.equals(execRes)) {
            return false;
        } else if (CACHE_NOT_ENOUGH.equals(execRes)) {
            throw new BusinessException("库存不足！");
        } else if (CACHE_NOT_EXIST.equals(execRes)) {
            // 库存不存在，初始化redis库存量
            Integer stockVal = initProductStockCache(spuId, skuId);
            if (stockVal != null) {
                incrementStock(spuId, skuId, num);
            } else {
                throw new BusinessException("库存扣减异常: 库存数据初始化异常");
            }
        }
        return false;
    }

    /**
     * redis 缓存rollback
     * @param spuId
     * @param skuId
     * @param num
     * @return
     */
    public boolean rollbackRedisStock(Long spuId, Long skuId, Integer num) {
        String stockRedisCacheKey = getProductStockCacheKey(spuId);
        String execRes = redisService.executeScript(RedisScript.of(LUAScriptConstants.UPDATE_SKU_STOCK_LUA, String.class),
                List.of(stockRedisCacheKey),
                String.valueOf(skuId), String.valueOf(-1 * num));
        return CACHE_UPDATE_SUCCESS.equals(execRes);
    }


    /**
     * 将库存更新任务丢到普通MQ，排队消费更新数据库（延迟更新，保证数据库与redis库存量最终一致）
     */
    private void pushProductStockUpdateMessageQueue(MallProductStockUpdateMessageDTO message ) {

        messageQueueProducer.send(MessageQueueEnum.QUEUE_MALL_PRODUCT_UPDATE_STOCK_MESSAGE, message);
    }

    /**
     * 构造数据库库存扣减队列消息
     *
     * @param skuId 商品SKU
     * @param num   更新库存量（正数代表回滚任务，负数代表扣减库存任务）
     * @return
     */
    private MallProductStockUpdateMessageDTO buildQueueMessage(Long skuId, Integer num) {
        MallProductStockUpdateMessageDTO message = new MallProductStockUpdateMessageDTO();
        message.setSkuId(skuId);
        message.setUpdateStock(num);
        return message;
    }

    private MessageQueueTaskDTO buildMessageQueueTaskDTO(MallProductStockUpdateMessageDTO message) {
        MessageQueueTaskDTO task = new MessageQueueTaskDTO();
        task.setTaskId(message.getTaskId());
        task.setTaskJson(JsonUtil.toJsonString(message));
        task.setStatus(MessageQueueTaskStatusEnum.ERROR.getStatus());
        task.setMethod(MessageQueueMethodEnum.SEND.getMethod());
        task.setCreateDate(new Date());
        task.setTopicName(MessageQueueEnum.QUEUE_MALL_PRODUCT_UPDATE_STOCK_MESSAGE.getQueue());
        return task;
    }

    /**
     * sku数据库库存量更新（库存更新队列consumer处理方法）
     *
     * @param messageDTO
     * @return
     */
    public int updateSkuStock(MallProductStockUpdateMessageDTO messageDTO) {

        return mallSkuDAO.updateSkuStock(messageDTO.getSkuId(), messageDTO.getUpdateStock());
    }

    /**
     * 查询单个SKU 库存
     *
     * 上游使用注意事项：剩余库存量可能为null（代表sku不存在）
     *
     * @param skuId
     * @return
     */
    public Integer getStock(Long spuId, Long skuId) {

        String cacheKey = getProductStockCacheKey(spuId);
        String stockCache = redisService.hget(cacheKey, String.valueOf(skuId));
        if (StringUtils.isNotBlank(stockCache)) {
            return Integer.parseInt(stockCache);
        }
        return initProductStockCache(spuId, skuId);
    }

    /**
     * 查询单个SKU 库存
     *
     * 上游使用注意事项：剩余库存量可能为null（代表sku不存在）
     *
     * @param skuIdList
     * @return
     */
    public Map<Long, Integer> batchGetStock(Long spuId, List<Long> skuIdList) {

        Map<Long, Integer> stockMap = new HashMap<>(skuIdList.size());

        String cacheKey = getProductStockCacheKey(spuId);
        List<String> fields = skuIdList.stream().map(String::valueOf).toList();
        List<String> stocks = redisService.hmget(cacheKey, fields);
        for (int i = 0; i < skuIdList.size(); i++) {
            Long skuId = skuIdList.get(i);
            String stockCache = stocks.get(i);
            Integer stockVal = null;
            if (StringUtils.isNotBlank(stockCache)) {
                stockVal = Integer.valueOf(stockCache);
            } else {
                stockVal = initProductStockCache(spuId, skuId);
            }
            stockMap.put(skuId, stockVal);
        }
        return stockMap;
    }


    private String getProductStockCacheKey(Long spuId) {

        return RedisKeyEnum.MALL_PRODUCT_SKU_STOCK_HASH_CACHE.formatKey(spuId);
    }

    /**
     * 初始化商品库存缓存
     *
     * @param spuId
     * @param skuId
     */
    private Integer initProductStockCache(Long spuId, Long skuId) {

        String refreshStockLock = RedisKeyEnum.MALL_REFRESH_PRODUCT_SKU_STOCK_LOCK.formatKey(spuId);
        RLock rLock = redissonClient.getLock(refreshStockLock);
        rLock.lock();
        try {
            // 双重检查锁，保证多分布式下只有单个线程完成redis中库存量的初始化
            String stockCacheKey = getProductStockCacheKey(spuId);
            String stockCache = redisService.hget(stockCacheKey, String.valueOf(skuId));
            if (StringUtils.isNotBlank(stockCache)) {
                return Integer.valueOf(stockCache);
            }
            // 数据库库存读取
            MallSKU sku = mallSkuDAO.findById(skuId).orElse(null);
            int stock = sku == null ? 0 : sku.getStock();
            // redis 库存写
            redisService.hset(stockCacheKey, String.valueOf(skuId), String.valueOf(stock),
                    RedisKeyEnum.MALL_PRODUCT_SKU_STOCK_HASH_CACHE.getExpireTime());
            return stock;
        } catch (Exception e) {

        } finally {
            rLock.unlock();
        }
        return null;
    }

}
